package com.xiaoshuidi.cloud.module.finance.mq.consumer;

import com.xiaoshuidi.cloud.framework.common.core.KeyValue;
import com.xiaoshuidi.cloud.framework.common.exception.enums.GlobalErrorCodeConstants;
import com.xiaoshuidi.cloud.framework.common.pojo.CommonResult;
import com.xiaoshuidi.cloud.framework.common.util.json.JsonUtils;
import com.xiaoshuidi.cloud.framework.tenant.core.util.TenantUtils;
import com.xiaoshuidi.cloud.module.contract.enums.contract.CustomerTypeEnum;
import com.xiaoshuidi.cloud.module.contract.vo.CustomerVO;
import com.xiaoshuidi.cloud.module.finance.enums.finance.FinanceCustomerTypeEnum;
import com.xiaoshuidi.cloud.module.finance.mq.message.FinanceAccountDeductMessage;
import com.xiaoshuidi.cloud.module.finance.service.finance.FinanceAccountService;
import io.netty.channel.ChannelOption;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.WriteTimeoutHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.netty.http.client.HttpClient;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
@Component("financeAccountDeduct")
public class FinanceAccountDeductConsumer implements Function<Flux<byte[]>, Mono<Void>> {
    private final FinanceAccountService financeAccountService;
    private final Sinks.Many<Collection<FinanceAccountDeductMessage>> sinks = Sinks.many().multicast().onBackpressureBuffer();
    private final WebClient webClient;
    private final Scheduler scheduler = Schedulers.newBoundedElastic(1,
            Runtime.getRuntime().availableProcessors(), "financeDeductScheduler", 300, true);

    public FinanceAccountDeductConsumer(FinanceAccountService financeAccountService,
                                        @Qualifier("loadBalancedWebClientBuilder") WebClient.Builder webClient) {
        this.financeAccountService = financeAccountService;
        HttpClient httpClient = HttpClient.create()
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)
                .doOnConnected(conn -> conn.addHandlerLast(new ReadTimeoutHandler(10))
                        .addHandlerLast(new WriteTimeoutHandler(10)));
        this.webClient = webClient.baseUrl("http://lease-contract-server")
                .clientConnector(new ReactorClientHttpConnector(httpClient))
                .build();
        failureListener();
    }

    private void failureListener() {
        subscribe(sinks.asFlux()
                .delayElements(Duration.ofSeconds(10))
                .flatMap(l -> {
                    List<FinanceAccountDeductMessage> retryMaxTimes = l.stream().filter(m -> m.getTimes() > 10).collect(Collectors.toList());
                    if (retryMaxTimes.size() > 0) {
                        log.info("账户扣费重试超过10次, 共{}条; [{}] 丢弃", retryMaxTimes.size(), retryMaxTimes.stream().map(FinanceAccountDeductMessage::getRoomId).collect(Collectors.toList()));
                        l.removeAll(retryMaxTimes);
                    }
                    if (l.size() > 0) {
                        return Flux.fromIterable(l.stream().collect(Collectors.groupingBy(FinanceAccountDeductMessage::getRoomId)).entrySet());
                    }
                    return Flux.empty();
                }))
                .doOnEach(s -> log.info("消费重试"))
                .subscribe();
    }

    /**
     * string字符串接收时, rocketmq默认的消息转换器MappingFastJsonMessageConverter会打印异常日志,
     * 虽然后面的MappingJackson2MessageConverter可以正常处理.
     * 加上以下配置对payload是string时不起作用, MappingFastJsonMessageConverter 定义又去不掉, 所以只能用Flux接收byte[]数组才不会进行消息payload转换
     * <pre>
     * spring:
     *   cloud:
     *     stream:
     *       bindings:
     *         financeAccountDeduct-in-0:
     *           destination: FINANCE_ACCOUNT
     *           group: financeAccountConsumer
     *           consumer:
     *             useNativeDecoding: true
     * </pre>
     * @param msg the function argument
     * @return
     */
    @Override
    public Mono<Void> apply(Flux<byte[]> msg) {
        return subscribe(msg.buffer(Duration.ofSeconds(2))
                .flatMap(buf -> {
                    List<FinanceAccountDeductMessage> l = new ArrayList<>();
                    for (byte[] b : buf) {
                        try {
                            if (b[0] == '[' && b[b.length - 1] == ']') {
                                l.addAll(JsonUtils.parseArray(new String(b), FinanceAccountDeductMessage.class));
                            } else {
                                l.add(JsonUtils.parseObject(b, FinanceAccountDeductMessage.class));
                            }
                        } catch (Exception e) {
                            log.error("扣费消息json反序列化异常: {}", new String(b));
                        }
                    }
                    Set<Map.Entry<Long, List<FinanceAccountDeductMessage>>> e = l.stream()
                            .filter(m -> m.getRoomId() != null && m.getAmount() != null && m.getAmount().compareTo(BigDecimal.ZERO) > 0)
                            .collect(Collectors.groupingBy(FinanceAccountDeductMessage::getRoomId)).entrySet();
                    log.info("收到 {} 条账户扣费消息", l.size());
                    // 按房间id分组
                    return Flux.fromIterable(e);
                }));
    }

    private final ParameterizedTypeReference<CommonResult<CustomerVO>> prf =
            new ParameterizedTypeReference<CommonResult<CustomerVO>>() {};
    private Mono<Void> subscribe(Flux<Map.Entry<Long, List<FinanceAccountDeductMessage>>> flux) {
        // 响应式中使用阻塞式Feign请求会报错, 这里需要用响应式的webClient
        return flux
                // 通过房间id获取租客id与客户类型
                .flatMap(e -> webClient.get().uri("/rpc-api/contract/get-member-id-by-room-id?roomId={roomId}", e.getKey())
                        .exchangeToMono(r -> r.bodyToMono(prf))
                        .map(result -> new KeyValue<>(e, result))
                        .cache(Duration.ofSeconds(10))
                        .onErrorResume(ex -> Mono.just(new KeyValue<>(e, CommonResult.error(GlobalErrorCodeConstants.INTERNAL_SERVER_ERROR))))
                )
                // 使用独立线程池
                .publishOn(scheduler)
                .doOnNext(this::deduct)
                .then();
    }

    /**
     * 处理扣费数据
     * @param e key是&lt;roomId, List&lt;FinanceAccountDeductMessage&gt;&gt; value是&lt;租客id, 账户类型&gt;
     */
    private void deduct(KeyValue<Map.Entry<Long, List<FinanceAccountDeductMessage>>, CommonResult<CustomerVO>> e) {
        if (e.getValue().isError()) {
            log.error("获取合同租客id失败, {} {}", e.getValue().getCode(), e.getValue().getMsg());
            addFailure(e.getKey().getValue());
        } else {
            TenantUtils.executeIgnore(() -> {
                try {
                    CustomerVO data = e.getValue().getData();
                    // FIXME 测试用, 找不到租客id默认返回0
                    data = Optional.ofNullable(data)
                            .map(v -> {
                                v.setType(CustomerTypeEnum.PERSONAL.getCode().equals(v.getType())
                                        ? FinanceCustomerTypeEnum.INDIVIDUAL.name()
                                        : ("room".equals(v.getType()) ? FinanceCustomerTypeEnum.ROOM.name() : FinanceCustomerTypeEnum.COMPANY.name()));
                                return v;
                            })
                            .map(v -> {
                                // 错误数据
                                if (v.getId() == null) {
                                    v.setId(0L);
                                    v.setType(FinanceCustomerTypeEnum.INDIVIDUAL.name());
                                    v.setTenantId(1L);
                                }
                                return v;
                            })
                            .orElse(CustomerVO.builder().id(0L).tenantId(1L).type(FinanceCustomerTypeEnum.INDIVIDUAL.name()).build());
                    String custId = FinanceCustomerTypeEnum.ROOM.name().equals(data.getType()) ? String.format("%d_%d", data.getCompanyId(), data.getRoomId())
                            : data.getId().toString();
                    String custType = data.getType();
                    Long roomId = e.getKey().getKey();
                    Long tenantId = data.getTenantId();
                    List<FinanceAccountDeductMessage> msgs = e.getKey().getValue();
                    if (!financeAccountService.deduct(custId, custType, roomId, data.getRoomName(), tenantId, msgs)) {
                        addFailure(msgs);
                    }
                } catch (Exception ex) {
                    log.error("账户扣减失败", ex);
                    addFailure(e.getKey().getValue());
                }
            });
        }
    }

    /**
     * 扣费失败后重新处理
     * @param msg
     */
    private void addFailure(Collection<FinanceAccountDeductMessage> msg) {
        msg.forEach(m -> m.setTimes(m.getTimes()+1));
        sinks.tryEmitNext(msg);
    }
}
